package com.ass.rocketmq.producer;

import com.alibaba.fastjson.JSON;
import com.ass.rocketmq.constant.MQTopic;
import com.ass.rocketmq.utils.MessageUtil;
import com.ass.rocketmq.base.AbstractTransactionListener;
import lombok.extern.log4j.Log4j2;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.event.ApplicationPreparedEvent;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Component;

@Component
@Log4j2
public abstract class AbstractTransactionProducer<M> {


    @Value("${spring.profiles.active}")
    private String active;

    @Value("${rocketmq.accessKey}")
    private String accessKey;

    @Value("${rocketmq.secretKey}")
    private String secretKey;

    @Value("${rocketmq.namespace}")
    private String namespace;


    @Value("${rocketmq.name-server}")
    private String namesrvAddr;

    private TransactionMQProducer transactionMQProducer;


    protected abstract MQTopic getMqTopic();

    //因为每个子类都要启动一个对应的生产者 所以要给一个实例名称
    protected abstract String getInstanceName();


    protected abstract <Listener extends AbstractTransactionListener> Listener getListener();


    /**
     * 初始化mq生产者
     */
    private void initTransactionProducer() throws MQClientException {
        log.info("transactionMQProducer 正在创建---------------------------------------");
        MQTopic mqTopic = getMqTopic();
        String prefix = active + "_";
        transactionMQProducer = new TransactionMQProducer (namespace,
                prefix+mqTopic.tag,
                new AclClientRPCHook(new SessionCredentials(accessKey, secretKey)));
        transactionMQProducer.setNamesrvAddr(namesrvAddr);
        transactionMQProducer.setVipChannelEnabled(false);
        transactionMQProducer.setRetryTimesWhenSendAsyncFailed(10);
        transactionMQProducer.setInstanceName(getInstanceName());
        transactionMQProducer.setTransactionListener(getListener());
        transactionMQProducer.start();
        log.info("transactionMQProducer 启动成功---------------------------------------");
    }

    /**
     * 启动事务生产者 要在spring 容器启动后 进行一些监听 以遍可以从ioc容器中 拿到想要的bean，
     * 否则使用@PostConstruct 注解 因为此注解是在bean初始化的时候执行 有一些业务的bean还没被加载到ioc里面
     * 可能有一些bean 加载不到
     *
     * @param event
     */
    @EventListener(ApplicationPreparedEvent.class)
    public void startTransactionProducer(ApplicationPreparedEvent event) {
        //获取bean的方法
//        PushConsumerProperty property = event.getApplicationContext().getBean(PushConsumerProperty.class);
        try {
            initTransactionProducer();
        } catch (MQClientException e) {
            log.error("[事务生产者start异常], nameServerAddress={}", namesrvAddr);
//            System.exit(-1);
        }
    }


    /**
     * 发送消息 直接配置参数在枚举里面，对应各个属性，不要写多个send方法
     */
    public void send(String keys,M m) {
        try {
            MQTopic mqTopic = getMqTopic();
            Message message= MessageUtil.buildMessage(keys, JSON.toJSONString(m),mqTopic,active);
            SendResult sendResult = transactionMQProducer.sendMessageInTransaction(message, null);
            log.info("sendStatus：{}",sendResult.getSendStatus());
        } catch (Exception e) {
            log.error(e);
        }
    }
}
